package com.syf.starter.kafka;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.syf.core.engine.SyFrameContext;
import com.syf.core.server.SyServer;
import com.syf.starter.kafka.config.SyKafkaCustomerServerConfiguration;
import com.syf.starter.kafka.handler.KafkaMessageHandlerManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import javax.annotation.PreDestroy;
import java.time.Duration;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;

/**
 * Kafka消费者服务器
 */
@Slf4j
public class SyKafkaCustomerServer implements SyServer {

    private final SyKafkaCustomerServerConfiguration configuration;
    private final AtomicBoolean running = new AtomicBoolean(true);
    private ExecutorService executorService;
    private KafkaMessageHandlerManager handlerManager;

    public SyKafkaCustomerServer(SyKafkaCustomerServerConfiguration configuration, SyFrameContext syFrameContext) {
        this.configuration = configuration;
        this.handlerManager = new KafkaMessageHandlerManager(syFrameContext);
    }

    @Override
    public void startup(CountDownLatch latch) {
        try {
            // 启动消费者线程
            executorService = Executors.newSingleThreadExecutor();
            executorService.submit(this::consumeMessages);
            log.info("Kafka消费者已启动");

            // 通知调用方已经启动
            latch.countDown();
        } catch (Exception e) {
            log.error("Kafka消费者启动失败", e);
            throw new RuntimeException("Kafka消费者启动失败", e);
        }
    }

    private KafkaConsumer<String, String> createConsumer() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configuration.getBootstrapServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, configuration.getGroupId());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, configuration.getAutoReset());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, configuration.getEnableAutoCommit());
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, configuration.getAutoCommitInterval());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, configuration.getMaxPollRecords());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        return new KafkaConsumer<>(props);
    }

    private void consumeMessages() {
        KafkaConsumer<String, String> consumer = createConsumer();
        try {
            // 订阅所有需要处理的topics
            consumer.subscribe(handlerManager.getAllTopics());
            log.info("Kafka消费者已订阅topic，topic集合：{}", handlerManager.getAllTopics());

            while (running.get()) {
                // 创建计时器（立即开始计时）
                TimeInterval timer = DateUtil.timer();
                try {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(configuration.getPollInterval()));
                    for (ConsumerRecord<String, String> record : records) {
                        handlerManager.handleMessage(record);
                    }
                } catch (Exception e) {
                    log.error("消费Kafka消息时发生错误", e);
                }

                // 计算已用时间（毫秒）
                long elapsed = timer.intervalMs();
                // 计算需要补充的休眠时间
                long sleepTime = configuration.getPollInterval() - elapsed - 10;

                // 如果处理时间不足configuration.getPollInterval()毫秒，则休眠剩余时间
                if (sleepTime > 0) {
                    try {
                        Thread.sleep(sleepTime);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        log.error("线程休眠被中断", e);
                    }
                }
                // 如果已超时（elapsed >= 1000），则直接进入下一轮循环
            }
        } finally {
            consumer.close();
        }
    }

    @Override
    public void gracefullyShutdown() {
        log.info("开始优雅关闭Kafka消费者...");
        // 设置停止标志，但给消费者一些时间处理已拉取的消息
        running.set(false);

        // 等待一段时间，确保已拉取的消息能够被处理
        try {
            log.info("等待2秒，确保已拉取的消息能够被处理...");
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            log.warn("等待过程中被中断", e);
            Thread.currentThread().interrupt();
        }

        // 关闭线程池
        if (executorService != null) {
            log.info("关闭Kafka消费者线程池...");
            executorService.shutdown();
            try {
                // 等待线程池中的任务完成，最多等待5秒
                if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    log.warn("Kafka消费者线程池未能在5秒内关闭，将强制关闭");
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                log.warn("等待线程池关闭过程中被中断", e);
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }

        log.info("Kafka消费者已关闭");
    }
}